package com.yonyou.common.service.messagecenter.rabbitmq.service;

import java.sql.Date;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.yonyou.common.service.messagecenter.config.RabbitMQConfig;
import com.yonyou.common.service.messagecenter.dao.TtMessageHistoryMapper;
import com.yonyou.common.service.messagecenter.rabbitmq.MessageProducer;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.TtMessageHistory;
import com.yonyou.common.service.messagecenter.util.FixedQueue;
import com.yonyou.common.service.messagecenter.util.MqConsts;

@Service
public class MessageHistorySyncService implements ITaskInfo{

	@Autowired
	private TtMessageHistoryMapper historyMapper;
	@Autowired
	private RabbitMQConfig config;
	@Autowired
	private MessageHistoryService messageHistoryService;
	private CopyOnWriteArrayList<TtMessageHistory> recordQueue = new CopyOnWriteArrayList<TtMessageHistory>();
	private CopyOnWriteArrayList<Integer> infos = new CopyOnWriteArrayList<Integer>();
	private DbRunnable runnable=new DbRunnable(this);
	private Thread my;
	private ConcurrentLinkedQueue counts;

	public MessageHistorySyncService(){
		my = new Thread(runnable); 
        //my.start(); 
	}
    @Transactional(rollbackFor=Exception.class)//annotationDrivenTransactionManager
    public  void insertHistory(TtMessageHistory record){
    	recordQueue.add(record);
    	infos.add(0);
//    	queue.add();
//    	DbRunnable runnable = new DbRunnable();
    	runnable.setHistoryMapper(this.historyMapper);
    	runnable.setRecordQueue(recordQueue);
    	runnable.sleeptime=config.getSleeptime();
    	runnable.messageHistoryService=this.messageHistoryService;
    	if(!my.isAlive()){
    		try{
                my.start(); 
    		}catch(Exception e){
    			
    		}
    	}
//    	runnable.setRecord(record);
//    	Thread my = new Thread(runnable); 
//    	my.setPriority(10);
//    	my.start();
    }
	@Override
	public void clear() {
		infos.clear();
	}
	@Override
	public boolean isBusy() {
		if(infos.size()>0)
			return true;
		return false;
	}
	@Override
	public void newThread() {
		my = new Thread(runnable); 
        my.start();
	}
}

class DbRunnable implements Runnable{

	private static Logger logger = LoggerFactory.getLogger(DbRunnable.class);
	private TtMessageHistoryMapper historyMapper;
	private TtMessageHistory record;
	protected MessageHistoryService messageHistoryService;
	private CopyOnWriteArrayList<TtMessageHistory> recordQueue;
	private ITaskInfo infos;
	protected int sleeptime;
	public DbRunnable(ITaskInfo infos) {
		 this.infos=infos;
	 }
	 
	@Override
	public void run() {
		double count=20;
		List<TtMessageHistory> list=new ArrayList();
		while (true){
			try {
				if(recordQueue!=null)
				{
					for(int i=1; i<=100;i++){
						if(recordQueue.size()>0){
							record=recordQueue.get(0);
							list.add(record);
							recordQueue.remove(0);
						}
					}
				}
				if(record!=null){
					messageHistoryService.insertBatch(list);
				}
//					historyMapper.insertBatch(list);//insert(record);//
				record=null;
				if(recordQueue!=null){
					if(recordQueue.size()>0 && !infos.isBusy()){
						count=1;
					}else{
						double c=recordQueue.size()/100F;
						count=c>sleeptime?sleeptime:c;//最多sleep 100秒
					}
					//当前线程停止运行，创建新线程
					if(recordQueue.size()==0){
						infos.newThread();
						break;
					}
				}
				count=count>0?count:20;
				list.clear();
				infos.clear();
				logger.info("---sleep time="+count*1000);
				Thread.sleep((int)count*1000l);
		} catch (Exception e) {
			logger.error(e.getMessage());
			list.clear();
		}
	}
	 }

	public TtMessageHistoryMapper getHistoryMapper() {
		return historyMapper;
	}

	public void setHistoryMapper(TtMessageHistoryMapper historyMapper) {
		this.historyMapper = historyMapper;
	}

	public TtMessageHistory getRecord() {
		return record;
	}

	public void setRecord(TtMessageHistory record) {
		this.record = record;
	}

	public CopyOnWriteArrayList<TtMessageHistory> getRecordQueue() {
		return recordQueue;
	}

	public void setRecordQueue(CopyOnWriteArrayList<TtMessageHistory> recordQueue) {
		this.recordQueue = recordQueue;
	}
	
}
	

